/*
 * Copyright (C) 2008 Universidade Federal de Campina Grande
 *  
 * This file is part of OurGrid. 
 *
 * OurGrid is free software: you can redistribute it and/or modify it under the
 * terms of the GNU Lesser General Public License as published by the Free 
 * Software Foundation, either version 3 of the License, or (at your option) 
 * any later version. 
 * 
 * This program is distributed in the hope that it will be useful, but WITHOUT 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
 * for more details. 
 * 
 * You should have received a copy of the GNU Lesser General Public License 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 * 
 */
package org.ourgrid.broker.controller;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.ourgrid.broker.BrokerConfiguration;
import org.ourgrid.broker.BrokerConstants;
import org.ourgrid.broker.actions.SchedulerAction;
import org.ourgrid.broker.commands.CommandExecutor;
import org.ourgrid.broker.commands.SchedulerData;
import org.ourgrid.broker.controller.messages.BrokerControlMessages;
import org.ourgrid.broker.controller.processors.ErrorOcurredProcessor;
import org.ourgrid.broker.controller.processors.HereIsFileInfoProcessor;
import org.ourgrid.broker.controller.processors.HereIsGridProcessResultProcessor;
import org.ourgrid.broker.controller.processors.WorkerIsReadyProcessor;
import org.ourgrid.broker.dao.BrokerMessageProcessorDAO;
import org.ourgrid.broker.dao.JobDAO;
import org.ourgrid.broker.dao.PeerDAO;
import org.ourgrid.broker.dao.PeerEntry;
import org.ourgrid.broker.dao.WorkerDAO;
import org.ourgrid.broker.scheduler.SchedulerIF;
import org.ourgrid.broker.scheduler.workqueue.WorkQueueReplication;
import org.ourgrid.broker.status.JobStatusInfo;
import org.ourgrid.broker.status.JobWorkerStatus;
import org.ourgrid.broker.status.PeerStatusInfo;
import org.ourgrid.broker.status.WorkerStatusInfo;
import org.ourgrid.common.interfaces.LocalWorkerProvider;
import org.ourgrid.common.interfaces.control.BrokerControlClient;
import org.ourgrid.common.interfaces.management.BrokerManager;
import org.ourgrid.common.interfaces.status.BrokerStatusProviderClient;
import org.ourgrid.common.interfaces.to.BrokerCompleteStatus;
import org.ourgrid.common.interfaces.to.JobEndedInterested;
import org.ourgrid.common.interfaces.to.JobsPackage;
import org.ourgrid.common.interfaces.to.PeersPackage;
import org.ourgrid.common.interfaces.to.WorkersPackage;
import org.ourgrid.common.job.JobCounter;
import org.ourgrid.common.job.PersistentJobCounter;
import org.ourgrid.common.job.SimpleJobCounter;
import org.ourgrid.common.spec.job.JobSpec;
import org.ourgrid.common.spec.peer.PeerSpec;
import org.ourgrid.peer.PeerConstants;
import org.ourgrid.reqtrace.Req;

import br.edu.ufcg.lsd.commune.Application;
import br.edu.ufcg.lsd.commune.api.FailureNotification;
import br.edu.ufcg.lsd.commune.api.MonitoredBy;
import br.edu.ufcg.lsd.commune.api.RecoveryNotification;
import br.edu.ufcg.lsd.commune.container.Container;
import br.edu.ufcg.lsd.commune.container.control.ApplicationControlClient;
import br.edu.ufcg.lsd.commune.container.control.ApplicationServerController;
import br.edu.ufcg.lsd.commune.container.control.ControlOperationResult;
import br.edu.ufcg.lsd.commune.container.servicemanager.dao.DAO;
import br.edu.ufcg.lsd.commune.identification.ContainerID;
import br.edu.ufcg.lsd.commune.identification.DeploymentID;
import br.edu.ufcg.lsd.commune.identification.ServiceID;

/**
 *
 */
public class BrokerComponentController extends ApplicationServerController implements BrokerManager {

	private JobCounter jobCounter;
	
	@Override
	protected void startComponent() throws Exception {
		configRepetitionActions();
		this.jobCounter = createJobCounter();
		
		initSchedulers();
		
		List<SchedulerData> outputCommands = new ArrayList<SchedulerData>();
		for(SchedulerIF scheduler: getJobDAO().getSchedulers()) {
			List<SchedulerData> cjData = scheduler.start();
			if (cjData != null) {
				outputCommands.addAll(cjData);
			}	
		}
		
		CommandExecutor.getInstance().execute(outputCommands, getServiceManager());
		
		getServiceManager().getLog().info(BrokerControlMessages.getSuccessfullyStartedBrokerMessage());
	}
	
	@Override
	protected boolean validateStartSenderPublicKey(ApplicationControlClient client, String senderPublicKey) {
		
		if(!getServiceManager().isThisMyPublicKey(senderPublicKey)) {
			getServiceManager().getLog().warn(BrokerControlMessages.getUnknownSenderControllingBrokerMessage(senderPublicKey));
			return false;
		}
		return true;
	}
	
	@Override
	protected boolean validateStopSenderPublicKey(ApplicationControlClient client, String senderPublicKey) {
		
		if(!getServiceManager().isThisMyPublicKey(senderPublicKey)) {
			getServiceManager().getLog().warn(BrokerControlMessages.getUnknownSenderControllingBrokerMessage(senderPublicKey));
			return false;
		}
		return true;
	}
	
	@Override
	protected void createServices() {
		getServiceManager().deploy(BrokerConstants.LOCAL_WORKER_PROVIDER_CLIENT, new LocalWorkerProviderClientController());
		getServiceManager().deploy(BrokerConstants.WORKER_CLIENT, new WorkerClientController());
		createMessageProcessors();
	}
	
	private BrokerMessageProcessorDAO getBrokerMessageProcessorDAO() {
		return getDAO(BrokerMessageProcessorDAO.class);
	}
	
	private <T extends DAO> T getDAO(Class<T> class1) {
		return getServiceManager().getDAO(class1);
	}
	
	private void createMessageProcessors() {
		getBrokerMessageProcessorDAO().
			putMessageProcessor(BrokerConstants.WORKER_IS_READY_ACTION_NAME, new WorkerIsReadyProcessor());
		getBrokerMessageProcessorDAO().
			putMessageProcessor(BrokerConstants.HERE_IS_EXECUTION_RESULT_ACTION_NAME, new HereIsGridProcessResultProcessor());
		getBrokerMessageProcessorDAO().
			putMessageProcessor(BrokerConstants.ERROR_OCURRED_ACTION_NAME, new ErrorOcurredProcessor());
		getBrokerMessageProcessorDAO().
			putMessageProcessor(BrokerConstants.HERE_IS_FILE_INFO_ACTION_NAME, new HereIsFileInfoProcessor());
	}
	
	@Override
	protected void createDAOs() {
		createDAO(JobDAO.class);
		createDAO(PeerDAO.class);
		createDAO(WorkerDAO.class);
		createDAO(BrokerMessageProcessorDAO.class);
	}
	
	private void createDAO(Class<? extends DAO> clazz) {
		getServiceManager().createDAO(clazz);
	}
	
	@Override
	protected void createMonitors() {
		getServiceManager().deploy(BrokerConstants.PEER_MONITOR_OBJECT_NAME, 
				new LocalWorkerProviderNotificationController());
		getServiceManager().deploy(BrokerConstants.WORKER_MONITOR_OBJECT_NAME, 
				new WorkerFailureController());
	}
	
	@Override
	public void stop(boolean callExit, boolean force, 
			@MonitoredBy(Application.CONTROL_OBJECT_NAME) ApplicationControlClient client) {
		
		String senderPublicKey = getServiceManager().getSenderPublicKey();
		
		if(!getServiceManager().isThisMyPublicKey(senderPublicKey)) {
			getServiceManager().getLog().warn(BrokerControlMessages.getUnknownSenderControllingBrokerMessage(senderPublicKey));
			return;
		}
		
		if (!canComponentBeUsed(client)) {
			getServiceManager().getLog().error(BrokerControlMessages.getComponentNotStartedMessage());
			return;
		}
		
		killJobDAO();
		finishRequests();
		
		List<SchedulerData> outputCommands = new ArrayList<SchedulerData>();
		for(SchedulerIF scheduler: getJobDAO().getSchedulers()) {
			List<SchedulerData> cjData = scheduler.stop();
			if (cjData != null) {
				outputCommands.addAll(cjData);
			}	
		}
		
		CommandExecutor.getInstance().execute(outputCommands, getServiceManager());
		
		getServiceManager().getLog().info(BrokerControlMessages.getSuccessfullyShutdownBrokerMessage());
		
		super.stop(callExit, force, client);
	}
	
	
	@Req("REQ304")
	public void addJob(@MonitoredBy(Application.CONTROL_OBJECT_NAME) BrokerControlClient brokerControlClient, JobSpec jobSpec) {
		
		String senderPublicKey = getServiceManager().getSenderPublicKey();
		
		if (!getServiceManager().isThisMyPublicKey(senderPublicKey)) {
			getServiceManager().getLog().warn(BrokerControlMessages.getUnknownSenderControllingBrokerMessage(senderPublicKey));
			return;
		}
		
		if (!canComponentBeUsed(brokerControlClient)) {
			getServiceManager().getLog().error(BrokerControlMessages.getComponentNotStartedMessage());
			return;
		}
		
		String maxReplicas = getServiceManager().getContainerContext().getProperty(BrokerConfiguration.PROP_MAX_REPLICAS);
		String maxFails = getServiceManager().getContainerContext().getProperty(BrokerConfiguration.PROP_MAX_FAILS);
		String maxBlFails = getServiceManager().getContainerContext().getProperty(BrokerConfiguration.PROP_MAX_BL_FAILS);
		
		int jobID = jobCounter.nextJobId();
		
		SchedulerIF scheduler = new WorkQueueReplication(Integer.valueOf(maxReplicas), 
				Integer.valueOf(maxFails), Integer.valueOf(maxBlFails));
		
		getJobDAO().setBrokerControlClient(brokerControlClient);
		getJobDAO().setScheduler(scheduler);
		getJobDAO().addJobSpec(jobID, jobSpec);
		
		List<SchedulerData> outputCommands = scheduler.addJob(jobSpec, jobID);
		CommandExecutor.getInstance().execute(outputCommands, getServiceManager());
	}

	@Req("REQ305")
	public void cancelJob(@MonitoredBy(Application.CONTROL_OBJECT_NAME)BrokerControlClient callback, int jobID) {
		
		String senderPubKey = getServiceManager().getSenderPublicKey();
		if (!getServiceManager().isThisMyPublicKey(senderPubKey)) {
			getServiceManager().getLog().warn(BrokerControlMessages.getUnknownSenderControllingBrokerMessage(senderPubKey));
			return;
		}
		
		if (!canComponentBeUsed(callback)) {
			getServiceManager().getLog().error(BrokerControlMessages.getComponentNotStartedMessage());
			return;
		}
		
		if (!getJobDAO().jobExists(jobID)) {
			getServiceManager().getLog().warn(BrokerControlMessages.getNoSuchJobToCancelMessage(jobID));
			callback.operationSucceed(new ControlOperationResult(
					new Exception(BrokerControlMessages.getNoSuchJobToCancelMessage(jobID))));
			return;
		}
		
		getJobDAO().setBrokerControlClient(callback);
		
		
		SchedulerIF scheduler = getJobDAO().getJobScheduler(jobID);
		List<SchedulerData> outputCommands = scheduler.cancelJob(jobID);
		CommandExecutor.getInstance().execute(outputCommands, getServiceManager());
	}

	@Req("REQ306")
	public void cleanAllFinishedJobs(@MonitoredBy(Application.CONTROL_OBJECT_NAME)BrokerControlClient callback) {
		
		String senderPubKey = getServiceManager().getSenderPublicKey();
		if (!getServiceManager().isThisMyPublicKey(senderPubKey)) {
			getServiceManager().getLog().warn(BrokerControlMessages.getUnknownSenderControllingBrokerMessage(senderPubKey));
			return;
		}
		
		if (!canComponentBeUsed(callback)) {
			getServiceManager().getLog().error(BrokerControlMessages.getComponentNotStartedMessage());
			return;
		}
		
		getJobDAO().setBrokerControlClient(callback);
		
		List<SchedulerData> outputCommands = new ArrayList<SchedulerData>();
		
		for(SchedulerIF scheduler: getJobDAO().getSchedulers()) {
			List<SchedulerData> cjData = scheduler.cleanAllFinishedJobs();
			if (cjData != null) {
				outputCommands.addAll(cjData);
			}	
		}
		
		CommandExecutor.getInstance().execute(outputCommands, getServiceManager());
	}

	@Req("REQ307")
	public void cleanFinishedJob(@MonitoredBy(Application.CONTROL_OBJECT_NAME)BrokerControlClient callback, int jobID) {
		
		String senderPubKey = getServiceManager().getSenderPublicKey();
		if (!getServiceManager().isThisMyPublicKey(senderPubKey)) {
			getServiceManager().getLog().warn(BrokerControlMessages.getUnknownSenderControllingBrokerMessage(senderPubKey));
			return;
		}
		
		if (!canComponentBeUsed(callback)) {
			getServiceManager().getLog().error(BrokerControlMessages.getComponentNotStartedMessage());
			return;
		}
		
		getJobDAO().setBrokerControlClient(callback);
		
		SchedulerIF scheduler = getJobDAO().getJobScheduler(jobID);
		List<SchedulerData> outputCommands = scheduler.cleanFinishedJob(jobID);
		CommandExecutor.getInstance().execute(outputCommands, getServiceManager());
	}
	
	@Req("REQ308")
	public void notifyWhenJobIsFinished(@MonitoredBy(Application.CONTROL_OBJECT_NAME)BrokerControlClient callback,
			@MonitoredBy(Application.CONTROL_OBJECT_NAME) JobEndedInterested interested, int jobID) {
		
		String senderPubKey = getServiceManager().getSenderPublicKey();
		if (!getServiceManager().isThisMyPublicKey(senderPubKey)) {
			getServiceManager().getLog().warn(BrokerControlMessages.getUnknownSenderControllingBrokerMessage(senderPubKey));
			return;
		}
		
		if (!canComponentBeUsed(callback)) {
			getServiceManager().getLog().error(BrokerControlMessages.getComponentNotStartedMessage());
			return;
		}
		
		getJobDAO().setBrokerControlClient(callback);
		
		DeploymentID stubDeploymentID = getServiceManager().getStubDeploymentID(interested);

		SchedulerIF scheduler = getJobDAO().getJobScheduler(jobID);
		List<SchedulerData> outputCommands = scheduler.notifyWhenJobIsFinished(jobID, stubDeploymentID.toString());
		CommandExecutor.getInstance().execute(outputCommands, getServiceManager());
	}

	@Req("REQ309")
	public void setPeers(@MonitoredBy(Application.CONTROL_OBJECT_NAME)BrokerControlClient callback, List<PeerSpec> peers) {
		
		String senderPubKey = getServiceManager().getSenderPublicKey();
		if (!getServiceManager().isThisMyPublicKey(senderPubKey)) {
			getServiceManager().getLog().warn(BrokerControlMessages.getUnknownSenderControllingBrokerMessage(senderPubKey));
			return;
		}
		
		if (!canComponentBeUsed(callback)) {
			getServiceManager().getLog().error(BrokerControlMessages.getComponentNotStartedMessage());
			return;
		}
		
		Collection<ContainerID> oldPeersIDs = getPeerDAO().getPeersIDs();
		
		String[] newPeerContainerID = new String[peers.size()];
		int i = 0;
		
		for (PeerSpec peerSpec : peers) {
			PeerEntry peerEntry = getPeerDAO().getPeerEntry(peerSpec.getServiceID().getContainerID());
			newPeerContainerID[i] = peerSpec.getServiceID().getContainerID().toString();
			i++;
			
			if (peerEntry == null) {
				getServiceManager().registerInterest(BrokerConstants.PEER_MONITOR_OBJECT_NAME, 
						peerSpec.getServiceID().toString(), 
						LocalWorkerProvider.class);
			} else {
				oldPeersIDs.remove(peerSpec.getServiceID().getContainerID());
			}
		}
		
		//unregister the remaining peers
		for (ContainerID peerID : oldPeersIDs) {
			LocalWorkerProvider localWorkerProvider = getPeerDAO().getPeerEntry(peerID).getLocalWorkerProvider();
			
			if (localWorkerProvider == null) {
				ServiceID peerServiceID = new ServiceID(peerID, PeerConstants.LOCAL_ACCESS_OBJECT_NAME);
				Container appContainer = getServiceManager().getApplication().getContainer();
				
				localWorkerProvider = (LocalWorkerProvider) appContainer.createStub(
						peerServiceID, LocalWorkerProvider.class);
			}

			finishRequests(peerID);
			getServiceManager().release(localWorkerProvider);
		}
		
		getPeerDAO().setPeers(peers);
		
		List<SchedulerData> outputCommands = new ArrayList<SchedulerData>();
		
		for(SchedulerIF scheduler: getJobDAO().getSchedulers()) {
			List<SchedulerData> cjData = scheduler.setPeers(newPeerContainerID);
			if (cjData != null) {
				outputCommands.addAll(cjData);
			}	
		}
		
		CommandExecutor.getInstance().execute(outputCommands, getServiceManager());
		
		callback.operationSucceed(new ControlOperationResult());
		
	}

	private void finishRequests(ContainerID peerID) {
		
		List<SchedulerData> outputCommands = new ArrayList<SchedulerData>();
		
		for(SchedulerIF scheduler: getJobDAO().getSchedulers()) {
			List<SchedulerData> cjData = scheduler.finishPeerRequests(peerID.toString());
			if (cjData != null) {
				outputCommands.addAll(cjData);
			}	
		}
		
		CommandExecutor.getInstance().execute(outputCommands, getServiceManager());
	}
	
	protected void finishRequests() {
		
		List<SchedulerData> outputCommands = new ArrayList<SchedulerData>();
		
		for(SchedulerIF scheduler: getJobDAO().getSchedulers()) {
			List<SchedulerData> cjData = scheduler.finishRequests();
			if (cjData != null) {
				outputCommands.addAll(cjData);
			}	
		}
		
		CommandExecutor.getInstance().execute(outputCommands, getServiceManager());
	}
	
	private void configRepetitionActions() {
		getServiceManager().addActionForRepetition(BrokerConstants.SCHEDULER_ACTION_NAME, new SchedulerAction());
	}
	
	/**
	 * Shuts the JobCounter down, by saving its last jobID into a file
	 */
	private void killJobDAO() {
		this.jobCounter.shutdown(true);				
	}

	/**
	 * Factory method for the JobCounter.
	 * @return JobCounter based on the Broker Context
	 */
	private JobCounter createJobCounter() {
		
		if (getServiceManager().getContainerContext().isEnabled(BrokerConfiguration.PROP_PERSISTJOBID)) {
			try {
				return new PersistentJobCounter(
						getServiceManager().getContainerContext().getProperty(BrokerConfiguration.PROP_JOBCOUNTERFILEPATH));
			} catch (IOException e) { }
		}
		return new SimpleJobCounter();
	}

	/**
	 * @return
	 */
	private PeerDAO getPeerDAO() {
		return getServiceManager().getDAO(PeerDAO.class);
	}
	
	private JobDAO getJobDAO() {
		return getServiceManager().getDAO(JobDAO.class);
	}
	
	@Override
	public String getComponentName() {
		return "Broker";
	}
	
	/**
	 * Retrieves to the callback client complete info on this Broker
	 * @param client The client that requested info
	 * @param entityID The entityID of this Broker status provider
	 */
	public void getCompleteStatus(@MonitoredBy(Application.CONTROL_OBJECT_NAME) BrokerStatusProviderClient client) {
		if (canStatusBeUsed()) {
			client.hereIsCompleteStatus( getServiceManager().getMyDeploymentID().getServiceID(), getCompleteStatus() );
		}
	}
	
	/**
	 * Retrieves to the callback client complete info on this Broker
	 * @param client The client that requested info
	 * @param entityID The entityID of this Broker status provider
	 */
	public void getJobsStatus(@MonitoredBy(Application.CONTROL_OBJECT_NAME) BrokerStatusProviderClient client, List<Integer> jobsIds) {
		if (canStatusBeUsed()) {
			client.hereIsJobsStatus( getServiceManager().getMyDeploymentID().getServiceID(), getJobsStatus(jobsIds) );
		}
	}
	
	private void initSchedulers() {
		
		String maxReplicas = getServiceManager().getContainerContext().getProperty(
				BrokerConfiguration.PROP_MAX_REPLICAS);
		String maxFails = getServiceManager().getContainerContext().getProperty(
				BrokerConfiguration.PROP_MAX_FAILS);
		String maxBlFails = getServiceManager().getContainerContext().getProperty(
				BrokerConfiguration.PROP_MAX_BL_FAILS);
		
		SchedulerIF scheduler = new WorkQueueReplication(Integer.valueOf(maxReplicas), 
				Integer.valueOf(maxFails), Integer.valueOf(maxBlFails));
		
		getJobDAO().addScheduler(scheduler);
	}
	
	/**
	 * Broker Complete status request
	 * @param client
	 */
	public BrokerCompleteStatus getCompleteStatus() {
		
//		JobWorkerStatus jwStatus = null;
//		Map<Integer, JobStatusInfo> jobs = new LinkedHashMap<Integer, JobStatusInfo>();
//		Map<Integer, HashSet<WorkerStatusInfo>> workers = new LinkedHashMap<Integer, HashSet<WorkerStatusInfo>>();
//		
//		for(SchedulerIF scheduler: getJobDAO().getSchedulers()) {
//			jwStatus = scheduler.getCompleteStatus();
//			jobs.putAll(jwStatus.getJobs());
//			workers.putAll(convertWorkerStatus(jwStatus.getWorkers()));
//		}
//		
//		List<PeerEntry> peers = new ArrayList<PeerEntry>(getPeerDAO().getPeers());
//		List<PeerStatusInfo> peerInfoList = new ArrayList<PeerStatusInfo>();
//		
//		for (PeerEntry peerEntry : peers) {
//			peerInfoList.add(new PeerStatusInfo(peerEntry.getState().toString(), peerEntry.getPeerSpec()));
//		}
		
		BrokerCompleteStatus status = new BrokerCompleteStatus(
				new JobsPackage(getAllJobsInfo()), 
				new PeersPackage(getAllPeersInfo()), 
				new WorkersPackage(getAllWorkersInfo()), 
				getServiceManager().getContainerDAO().getUpTime(), 
				getServiceManager().getContainerContext().toString());
		
		return status;
	}
	
	public void registerAsListener(@MonitoredBy(Application.CONTROL_OBJECT_NAME) BrokerStatusProviderClient client) {
		// TODO Auto-generated method stub
	}
	
	@RecoveryNotification
	public void statusProviderClientIsUp(BrokerStatusProviderClient statusProviderClient) {}
	
	@FailureNotification
	public void statusProviderClientIsDown(BrokerStatusProviderClient statusProviderClient) {}
	
	@RecoveryNotification
	public void controlClientIsUp(BrokerControlClient client) {}
	
	@FailureNotification
	public void controlClientIsDown(BrokerControlClient client) {}
	
	@RecoveryNotification
	public void jobEndedInterestedIsUp(JobEndedInterested interested, DeploymentID interestedID){}
	
	@FailureNotification
	public void jobEndedInterestedIsDown(JobEndedInterested interested, DeploymentID interestedID){
		
		List<SchedulerData> outputCommands = new ArrayList<SchedulerData>();
		
		for(SchedulerIF scheduler: getJobDAO().getSchedulers()) {
			List<SchedulerData> cjData = scheduler.jobEndedInterestedIsDown(interestedID.toString());
			outputCommands.addAll(cjData);
		}
		
		CommandExecutor.getInstance().execute(outputCommands, getServiceManager());
		
		getServiceManager().release(interested);
	}
	
	private Map<Integer, HashSet<WorkerStatusInfo>> convertWorkerStatus(Map<Integer, WorkerStatusInfo[]> statusMap) {
		
		Map<Integer, HashSet<WorkerStatusInfo>> newMap = new LinkedHashMap<Integer, HashSet<WorkerStatusInfo>>();
		
		if (statusMap != null) {
			for (Entry<Integer, WorkerStatusInfo[]> status : statusMap.entrySet()) {
				newMap.put(status.getKey(), getSet(status.getValue()));
			}
		}
		
		return newMap;
	}
	
	private HashSet<WorkerStatusInfo> getSet(WorkerStatusInfo[] array) {
		
		HashSet<WorkerStatusInfo> set = new LinkedHashSet<WorkerStatusInfo>();
		
		if (array != null) {
			for (WorkerStatusInfo workerStatusInfo : array) {
				set.add(workerStatusInfo);
			}
		}
		
		return set;
	}
	
	private Map<Integer, JobStatusInfo> getAllJobsInfo() {
		JobWorkerStatus jwStatus = null;
		Map<Integer, JobStatusInfo> allJobs = new LinkedHashMap<Integer, JobStatusInfo>();
		Map<Integer, HashSet<WorkerStatusInfo>> allWorkers = new LinkedHashMap<Integer, HashSet<WorkerStatusInfo>>();
		
		for(SchedulerIF scheduler: getJobDAO().getSchedulers()) {
			jwStatus = scheduler.getCompleteStatus();
			allJobs.putAll(jwStatus.getJobs());
			allWorkers.putAll(convertWorkerStatus(jwStatus.getWorkers()));
		}
		return allJobs;
	}
	
	private JobsPackage getJobsStatus(List<Integer> jobsIds) {
		Map<Integer, JobStatusInfo> allJobsInfo = getAllJobsInfo();
		Map<Integer, JobStatusInfo> selectedJobsInfo = new HashMap<Integer, JobStatusInfo>();
		
		for (Integer jobId : jobsIds) {
			selectedJobsInfo.put(jobId, allJobsInfo.get(jobId));
		}
		
		return new JobsPackage(selectedJobsInfo);
	}
	
	private Map<Integer, HashSet<WorkerStatusInfo>> getAllWorkersInfo() {
		Map<Integer, HashSet<WorkerStatusInfo>> allWorkers = new LinkedHashMap<Integer, HashSet<WorkerStatusInfo>>();
		JobWorkerStatus jwStatus = null;
		
		for(SchedulerIF scheduler: getJobDAO().getSchedulers()) {
			jwStatus = scheduler.getCompleteStatus();
			allWorkers.putAll(convertWorkerStatus(jwStatus.getWorkers()));
		}
		return allWorkers;
	}
	
	private List<PeerStatusInfo> getAllPeersInfo() {
		List<PeerEntry> peers = new ArrayList<PeerEntry>(getPeerDAO().getPeers());
		List<PeerStatusInfo> peerInfoList = new ArrayList<PeerStatusInfo>();
		
		for (PeerEntry peerEntry : peers) {
			peerInfoList.add(new PeerStatusInfo(peerEntry.getState().toString(), peerEntry.getPeerSpec()));
		}
		return peerInfoList;
	}

}
